package kafka

import (
	"github.com/Shopify/sarama"
)

//var syncProducerList = make(map[string]*syncProducer)
//
////GetSyncProducerClient 从kafka配置项获取brokerConfKey对应的broker列表，返回同步生产者
//func GetSyncProducerClient(brokerConfKey string) (*syncProducer, error) {
//	if _, ok := syncProducerList[brokerConfKey]; !ok { //第一次来
//		brokerList := viper.GetStringSlice("kafka." + brokerConfKey + ".brokers")
//		if len(brokerList) <= 0 {
//			return nil, errors.New("在配置文件中未发现该项配置")
//		}
//
//		if producerClient, err := newSyncProducerClient(brokerList, func(config *sarama.Config) {
//			config.Producer.Flush.Frequency = 500 * time.Millisecond // 每0.5s冲洗一批
//
//			acks := viper.GetInt("kafka." + brokerConfKey + ".producer.acks")
//			config.Producer.RequiredAcks = sarama.RequiredAcks(acks)
//
//			config.Producer.Partitioner = sarama.NewRandomPartitioner // 选择分区（此处随机设置一个分区）
//		}); err == nil {
//			syncProducerList[brokerConfKey] = &syncProducer{Producer: producerClient}
//		} else {
//			return nil, err
//		}
//	}
//	return syncProducerList[brokerConfKey], nil
//}
//
//func DeleteSyncProducerClient(brokerConfKey string) {
//	if client, ok := syncProducerList[brokerConfKey]; ok {
//		_ = client.close()
//		delete(syncProducerList, brokerConfKey)
//	}
//}

// NewSyncProducerClient 同步生产者
func NewSyncProducerClient(brokers []string, options ...func(*sarama.Config)) (sarama.SyncProducer, error) {
	kafkaConfig := sarama.NewConfig()
	kafkaConfig.Producer.Return.Successes = true                // 成功交付的消息将在success channel返回
	kafkaConfig.Producer.Compression = sarama.CompressionSnappy // 压缩消息

	for _, option := range options {
		option(kafkaConfig)
	}

	// 连接kafka
	client, err := sarama.NewSyncProducer(brokers, kafkaConfig)
	if err != nil {
		return nil, err
	}

	return client, nil
}

//
//type syncProducer struct {
//	Producer sarama.SyncProducer
//}
//
//func (p *syncProducer) Send(topic string, value interface{}) (int32, int64, error) {
//	if p.Producer == nil {
//		return 0, 0, errors.New("Kafka producer is nil")
//	}
//
//	msg, err := json.Marshal(value)
//	if err != nil {
//		return 0, 0, err
//	}
//
//	// 发送消息
//	return p.Producer.SendMessage(&sarama.ProducerMessage{
//		Topic:     topic,
//		Value:     sarama.ByteEncoder(msg),
//		Timestamp: time.Now(),
//	})
//}
//
//func (p *syncProducer) close() error {
//	return p.Producer.Close()
//}
